package com.yanggu.spark.core.serialize

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Serialize_Fun02_Kryo {

  def main(args: Array[String]): Unit = {
    //1. 创建sparkConf
    val sparkConf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("KryoDemo")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .registerKryoClasses(Array(classOf[Searcher]))

    //2. 创建上下文环境对象
    val sparkContext = new SparkContext(sparkConf)

    val rdd: RDD[String] = sparkContext.makeRDD(Array("hello world", "hello atguigu", "atguigu", "hahah"), 2)

    val searcher = new Searcher("hello")
    val result: RDD[String] = searcher.getMatchedRDD1(rdd)

    result.collect.foreach(println)
  }

}
case class Searcher(val query: String) {

  def isMatch(s: String) = {
    s.contains(query)
  }

  def getMatchedRDD1(rdd: RDD[String]) = {
    rdd.filter(isMatch)
  }

  def getMatchedRDD2(rdd: RDD[String]) = {
    val q = query
    rdd.filter(_.contains(q))
  }
}